package com.gradle.consumer.kafka.config;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@EnableKafka
public class KafkaConsumer {
    /**
     * 监听消费
     * @param record record
     */
//    @KafkaListener(topics = {"consumer_topic"})
//    public void onNormalMessage(ConsumerRecord<String, Object> record) {
//        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +
//                record.value());
//    }



    @KafkaListener(id = "producer-1",topics = {"topic1001"})
    /**
     * 批量消费
     * id- 监听器id(可用来命名消费者线程)
     * containerFactory -  监听器工厂
     * idIsGroup - 是否使用id作为groupId
     * @param records
     */
    public void onBatchMessage(List<ConsumerRecord<String, Object>> records) {
        System.out.println(">>> 批量消费一次，recoreds.size()=" + records.size());
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println("消息内容为："+record.value());
        }
    }
}
